﻿using log4net.Appender;
using log4net.Core;
using log4net.Util;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;

namespace WL.Think.Log4net.AsyncBulkAppender
{
    public abstract class AsyncBulkAppenderSkeleton : AppenderSkeleton
    {
        /// <summary>
        /// 默认最大缓冲区大小
        /// </summary>
        private const int DefaultFlushTriggerBufferSize = 256;

        /// <summary>
        /// 滚动缓冲区的默认大小
        /// </summary>
        private const int DefaultRollingBufferSize = 50;

        /// <summary>
        /// 当所有事件都已处理且队列已关闭时触发
        /// </summary>
        private readonly ManualResetEvent addingCompletedEvent;

        /// <summary>
        /// 用于关闭事件队列的取消令牌源
        /// </summary>
        private readonly CancellationTokenSource cancellationTokenSource;

        /// <summary>
        /// 队列日志事件的集合
        /// </summary>
        private BlockingCollection<LoggingEvent> eventsQueue;


        /// <summary>
        /// 初始化
        /// </summary>
        protected AsyncBulkAppenderSkeleton() : this(DefaultRollingBufferSize, DefaultFlushTriggerBufferSize)
        {

        }


        /// <summary>
        /// 初始化
        /// </summary>
        /// <param name="rollingBufferSize"></param>
        /// <param name="flushTriggerBufferSize"></param>
        protected AsyncBulkAppenderSkeleton(int rollingBufferSize, int flushTriggerBufferSize)
        {
            RollingBufferSize = rollingBufferSize;
            FlushTriggerBufferSize = flushTriggerBufferSize;

            //信号发射器，true是发送信号(终止状态)，当有信号发送时，所有等待的线程都会继续执行
            //而false则表示未发出信号（非终止状态），那么所有WaitOne的线程都要等待，知道信号发出
            //终止状态时WaitOne()允许线程访问下边的语句
            //非终止状态时WaitOne()阻塞线程，不允许线程访问下边的语句
            //ManualResetEvent就像一个水龙头一样，只要开关一打开，所有的线程都可以执行，已关闭就要等待
            addingCompletedEvent = new ManualResetEvent(false);
            cancellationTokenSource = new CancellationTokenSource();
        }


        /// <summary>
        /// 获取或设置打开关闭超时
        /// </summary>
        public TimeSpan OnCloseTimeout { get; set; }


        /// <summary>
        /// 获取或设置日志事件的滚动缓冲区的最大大小
        /// </summary>
        public int RollingBufferSize { get; set; }


        /// <summary>
        /// 获取或设置将立即触发刷新日志事件的缓冲区的最大大小
        /// </summary>
        public int FlushTriggerBufferSize { get; set; }


        /// <inheritdoc />
        public override void ActivateOptions()
        {
            base.ActivateOptions();

            eventsQueue = RollingBufferSize > 0 ? new BlockingCollection<LoggingEvent>(RollingBufferSize) : new BlockingCollection<LoggingEvent>();

            //开启一个不关闭的任务
            Task.Factory.StartNew(LoggingWorker, TaskCreationOptions.LongRunning);
        }

        /// <summary>
        /// 向异步缓冲区追加新消息
        /// </summary>
        /// <param name="loggingEvent">Logging event with the message</param>
        protected override void Append(LoggingEvent loggingEvent)
        {
            try
            {
                loggingEvent.Fix = FixFlags.Message | FixFlags.Exception;

                //如果IsAddingCompleted = true，那么集合就不能再接受添加数据
                if (!eventsQueue.IsAddingCompleted)
                {
                    int droppedCount = 0;
                    //往集合中添加数据，如果数据已经达到最大容量，那么这里会显示添加失败
                    //可以配置第二个参数来表示如果阻塞，等待多久直接返回false
                    while (!eventsQueue.TryAdd(loggingEvent))
                    {
                        //从集合中读取一个数据，必须保证集合中有数据，否则会报错
                        eventsQueue.TryTake(out _);
                        droppedCount++;
                    }

                    if (droppedCount > 0)
                    {
                        LogLog.Warn(GetType(), $"{droppedCount} log events have been dropped, RollingBufferSize={RollingBufferSize} has been reached");
                    }
                }
            }
            catch (Exception ex)
            {
                ErrorHandler.Error("Exception occurred in AsyncAppender wrapper", ex);
            }
        }

        /// <summary>
        /// 关闭日志记录器时调用的操作
        /// </summary>
        protected override void OnClose()
        {
            eventsQueue.CompleteAdding();
            addingCompletedEvent.WaitOne(OnCloseTimeout);
            cancellationTokenSource.Cancel();
            base.OnClose(); 
        }

        /// <summary>
        /// 异步调用的方法，负责日志记录(提供要日志记录的事件集合)
        /// </summary>
        /// <param name="loggingEvents">Logging event with the message</param>
        protected abstract void AppendBulk(IList<LoggingEvent> loggingEvents);


        /// <summary>
        /// 作为消息队列的使用者线程的工作方法
        /// </summary>
        private void LoggingWorker()
        {
            try
            {
                var buffer = new List<LoggingEvent>();

                //GetConsumingEnumerable这个方法会实现获取集合，但是带阻塞的，如果集合中没有数据，会一直阻塞在这里
                //直到有数据后才会继续运行
                //另外，如果集合已经CompleteAdding() 也会退出循环
                foreach (var loggingEvent in eventsQueue.GetConsumingEnumerable(cancellationTokenSource.Token))
                {
                    buffer.Add(loggingEvent);
                    if (eventsQueue.Count == 0 || buffer.Count >= FlushTriggerBufferSize)
                    {
                        try
                        {
                            AppendBulk(buffer);
                        }
                        catch (Exception ex)
                        {
                            ErrorHandler.Error("Exception occurred in AsyncBulkAppenderSkeleton wrapper while adding current batch of events.", ex);
                        }
                        finally
                        {
                            buffer.Clear();
                        }
                    }
                }

                //释放信号，信号释放后，发出等待线程可以继续进行的信号。并释放所有等待线程
                //Reset() 则是重置为不释放信号，那么其他线程就得等待
                //把非终止状态改为终止状态用Set()方法
                //把终止状态改为非终止状态用Reset()方法
                addingCompletedEvent.Set();
            }
            catch (Exception ex)
            {
                ErrorHandler.Error("Exception occurred in AsyncBulkAppenderSkeleton main worker loop. No more events will be processed.", ex);
            }
        }
    }
}
